use std::{
    path::{Path, PathBuf},
    sync::Arc,
};

use crate::{
    app::{
        FileDownloadService, FileProcessResult, GrpcService, P2pCommand, P2pService,
        P2pServiceConfig, store::rocksdb::RocksDb,
    },
    error::ServerError,
};
use tokio::{
    fs,
    sync::{Mutex, mpsc},
    task::JoinSet,
};
use tokio_util::sync::CancellationToken;
use tracing::info;

const LOG_TARGET: &str = "app::server";

pub type ServerResult<T> = Result<T, ServerError>;

/// 目前，Rust 还没有稳定的 trait 异步化支持，原生的异步trait不支持动态分发
/// 因此，我们需要使用一个第三方库来实现异步trait
/// 这里我们使用 async_trait 库来实现异步trait
#[async_trait::async_trait]
pub trait Service: Send + Sync + 'static {
    async fn start(&mut self, cancel_token: CancellationToken) -> Result<(), ServerError>;
}

#[derive(Debug, Default)]
pub struct Server {
    cancel_token: CancellationToken,
    sub_tasks: Arc<Mutex<JoinSet<Result<(), ServerError>>>>,
    grpc_port: u16,
    base_path: PathBuf,
}

impl Server {
    pub fn new(grpc_port: u16, base_path: impl AsRef<Path>) -> Self {
        Self {
            cancel_token: CancellationToken::new(),
            sub_tasks: Arc::new(Mutex::new(JoinSet::new())),
            grpc_port,
            base_path: base_path.as_ref().to_path_buf(),
        }
    }

    pub async fn start(&mut self) -> ServerResult<()> {
        info!(target: LOG_TARGET, "Starting server...");
        self.init_base_dir().await?;
        let (file_publish_sender, file_publish_receiver) = mpsc::channel::<FileProcessResult>(100);
        let (p2p_command_sender, p2p_command_receiver) = mpsc::channel::<P2pCommand>(100);
        // p2p service

        let file_store = RocksDb::new(self.base_path.join("file_store"))?;
        let p2p_service = P2pService::new(
            P2pServiceConfig::builder()
                .with_keypair_file(self.base_path.join("keys.keypair"))
                .build(),
            file_publish_receiver,
            p2p_command_receiver,
            file_store.clone(),
        );
        self.spawn_task(p2p_service).await?;

        // grpc service
        let grpc_service = GrpcService::new(
            self.grpc_port,
            file_publish_sender,
            p2p_command_sender.clone(),
            file_store.clone(),
        );
        self.spawn_task(grpc_service).await?;

        // file download service
        let file_download_service = FileDownloadService::new(file_store, 10, p2p_command_sender);
        self.spawn_task(file_download_service).await?;

        info!(target: LOG_TARGET, "Server started.");
        Ok(())
    }

    pub async fn spawn_task<S: Service>(&mut self, mut service: S) -> ServerResult<()> {
        let mut join_set = self.sub_tasks.lock().await;
        let cancel_token = self.cancel_token.clone();
        join_set.spawn(async move { service.start(cancel_token).await });

        Ok(())
    }

    pub async fn stop(&self) -> ServerResult<()> {
        info!(target: LOG_TARGET, "Stopping server...");
        self.cancel_token.cancel();
        let mut join_set = self.sub_tasks.lock().await;
        while let Some(res) = join_set.join_next().await {
            res??;
        }
        info!(target: LOG_TARGET, "Server stopped.");
        Ok(())
    }

    async fn init_base_dir(&self) -> ServerResult<()> {
        // return dir if exists
        if let Ok(metadata) = fs::metadata(&self.base_path).await {
            if !metadata.is_dir() {
                return Err(ServerError::InitBaseDirError(self.base_path.clone()));
            } else {
                return Ok(());
            }
        }

        // create dir if not exists
        fs::create_dir_all(&self.base_path).await?;
        Ok(())
    }
}
